package com.innodata.riskEngine.utils;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @Author liboshuai
 * @Date 2023/11/2 17:19
 */
@Slf4j
@Component
public class KafkaTemplateUtil {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    public void sendMessage(String topic, Object message) {
        kafkaTemplate.send(topic, message);
        log.info("发送message >>> {}", message);
    }

    public void sendWithCallback(String topic, Object message) {
        kafkaTemplate.send(topic, message).addCallback(
                success -> {
                    String topicName = success.getRecordMetadata().topic();
                    int partition = success.getRecordMetadata().partition();
                    long offset = success.getRecordMetadata().offset();
                    Object value = success.getProducerRecord().value();
                    log.info("消息被发送到 topic >>> {}, 分区 >>> {}, offset >>> {}, 消息内容为 >>> {};", topicName, partition, offset, value);
                },
                failure -> log.info("消息发送失败 >>> {}; ", failure.getMessage()));
    }

    public void sendWithDetailedCallback(String topic, Object message) {
        kafkaTemplate.send(topic, message).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.info("消息发送失败 >>> {}; ", ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                String topicName = result.getRecordMetadata().topic();
                int partition = result.getRecordMetadata().partition();
                long offset = result.getRecordMetadata().offset();
                Object value = result.getProducerRecord().value();
//                log.info("消息被发送到 topic >>> {}, 分区 >>> {}, offset >>> {}, 消息内容为 >>> {};", topicName, partition, offset, value);
            }
        });
    }

    public void sendInTransactWithStop(String topic, Object message) {
        // 声明事务，如果出现异常，则消息不会继续发送
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send(topic, message);
            throw new RuntimeException("fail");
        });
    }

    public void sendInTransactWithContinue(String topic, Object message) {
        // 声明事务，如果出现异常，消息仍会继续发送
        kafkaTemplate.send(topic, message);
        throw new RuntimeException("fail");
    }
}